package com.li.flink.datastream.state

import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.time.Time
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import java.util.concurrent.TimeUnit


object CheckpointApp {
  def main(args: Array[String]): Unit = {
    //获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //开启checkpoint
    env.enableCheckpointing(5000)
    //new MemoryStateBackend(0, false);
    env.setStateBackend(new FsStateBackend("file:///Users/lijiacen/Downloads/backend", false))
    //保留checkoutpoint在磁盘
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    //重启策略
    //    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000))
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(10, TimeUnit.SECONDS)))
    test01(env)
    env.execute("CheckpointApp")
  }

  def test01(env: StreamExecutionEnvironment): Unit = {
    //连接socket
    var socketStream = env.socketTextStream("127.0.0.1", 9001)
    socketStream.print()
  }
}
